/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.commons.helper;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.ipig.commons.context.JavaContext;
import org.ipig.constants.SymbolCnst;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * 动态可伸缩线程池
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: ThreadHelper.java 15 2020-04-21 12:04:17Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
public class ThreadHelper {
    public static final int MAXINUM_AVAILABLE_PROCESSOR_SIZE = Runtime.getRuntime().availableProcessors();
    public static final int MAXINUM_POOL_SIZE = 128;
    private static final String THREAD_DEFAULT_NAME = "skeleton";
    private static int processorSize = MAXINUM_AVAILABLE_PROCESSOR_SIZE;
    private static int corePoolSize = MAXINUM_AVAILABLE_PROCESSOR_SIZE;
    private static int maxinumPoolSize = MAXINUM_AVAILABLE_PROCESSOR_SIZE;
    private static int queueCapacity = MAXINUM_AVAILABLE_PROCESSOR_SIZE;
    private static int waitMillis = 5;
    private static int keepAliveSeconds = 300;
    private static int continuedWaitingCycleCount = 400;//当waitMillis = 5时，2秒内都没新任务加入时Warn
    private static String threadName = THREAD_DEFAULT_NAME;
    private static volatile ExecutorService executorService = null;
    private static volatile BlockingQueue queue = null;
    private static volatile LongAdder threadNumber = new LongAdder();
    private static volatile boolean daemon = true;
    private static Properties props = new Properties();
    private static String configProps = "config.properties";
    ;

    static {
        String osName = JavaContext.getOsName();
        if (StringUtils.startsWithIgnoreCase(osName, "win") ||
                StringUtils.startsWithIgnoreCase(osName, "mac")) {
            if (MAXINUM_AVAILABLE_PROCESSOR_SIZE <= 4) {
                processorSize = 4;
                maxinumPoolSize = 16;
            } else if (MAXINUM_AVAILABLE_PROCESSOR_SIZE <= 8) {
                maxinumPoolSize = 32;
            } else {
                maxinumPoolSize = processorSize * 4;
            }
        } else {
            if (MAXINUM_AVAILABLE_PROCESSOR_SIZE <= 2) {
                processorSize = 4;
                maxinumPoolSize = 32;
            } else if (MAXINUM_AVAILABLE_PROCESSOR_SIZE <= 4) {
                processorSize = 8;
                maxinumPoolSize = 64;
            } else {
                processorSize = MAXINUM_AVAILABLE_PROCESSOR_SIZE * 2;
                maxinumPoolSize = processorSize * 16;
            }
        }
        corePoolSize = processorSize * 2;
        if (maxinumPoolSize > MAXINUM_POOL_SIZE) {
            maxinumPoolSize = MAXINUM_POOL_SIZE;
        }
        queueCapacity = processorSize * 4;
    }

    /**
     * getThreadExecutor
     *
     * @return
     */
    private static ExecutorService getThreadExecutor() {
        if (executorService == null) {
            synchronized (ThreadHelper.class) {
                if (executorService == null) {
                    ThreadFactory threadFactory = new ThreadFactory() {
                        @Override
                        public Thread newThread(Runnable runnable) {
                            threadNumber.increment();
                            Thread thread = new Thread(runnable, threadName + SymbolCnst.MIDLINE +
                                    threadNumber.longValue() + SymbolCnst.MIDLINE + DateTimeHelper.getDateTime());
                            thread.setDaemon(daemon);
                            return thread;
                        }
                    };
                    RejectedExecutionHandler rejectedHandler = new RejectedExecutionHandler() {
                        @Override
                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                            int depthQueue = 0, count = 0;
                            while ((depthQueue = executor.getQueue().size()) >= queueCapacity) {
                                try {
                                    Thread.sleep(waitMillis);
                                    if (count++ >= continuedWaitingCycleCount) {
                                        log.warn("{} thread pool's blocking queue depth is: {}, active threads ={}, " +
                                                        " pool size = {},task count = {}, [configuration info: " +
                                                        "corePoolSize={},maxinumPoolSize={},keepAliveSeconds={}]",
                                                threadName, depthQueue, executor.getActiveCount(), executor.getPoolSize(),
                                                executor.getTaskCount(), corePoolSize, maxinumPoolSize, keepAliveSeconds);
                                        count = 0;
                                    }
                                } catch (InterruptedException ex) {
                                    ex.printStackTrace();
                                }
                            }
                            executor.submit(r);
                        }
                    };

                    PropertiesHelper propsHelper = null;
                    try {
                        InputStream is = ClassLoaderHelper.getResourceAsStream(configProps, ThreadHelper.class);
                        if (is != null) {
                            props.load(is);
                            propsHelper = new PropertiesHelper(props);
                        }
                    } catch (Exception ex) {
                        System.out.println(ex.getMessage());
                    }
                    if (propsHelper != null) {
                        corePoolSize = propsHelper.getInt("thread.pool.core.size", corePoolSize);
                        maxinumPoolSize = propsHelper.getInt("thread.pool.max.size", maxinumPoolSize);
                        keepAliveSeconds = propsHelper.getInt("thread.pool.alive.seconds", keepAliveSeconds);
                        queueCapacity = propsHelper.getInt("thread.pool.queue.capacity", queueCapacity);
                    }
                    queue = new ArrayBlockingQueue(queueCapacity);
                    executorService = new ThreadPoolExecutor(maxinumPoolSize, maxinumPoolSize, keepAliveSeconds,
                            TimeUnit.SECONDS, queue, threadFactory, rejectedHandler);
/*                    executorService = new ThreadPoolExecutor(corePoolSize, maxinumPoolSize, keepAliveSeconds,
                            TimeUnit.SECONDS, queue, threadFactory, rejectedHandler);*/
                    Runtime.getRuntime().addShutdownHook(new Thread("application-shutdown-hook") {
                        @Override
                        public void run() {
                            if (executorService != null && !executorService.isShutdown()) {
                                executorService.shutdown();
                            }
                        }
                    });
                }
            }
        }
        return executorService;
    }

    /**
     * execute
     * <ul>注：不能保持执行顺序</ul>
     *
     * @param task
     */
    public static <T> Future<T> execute(Callable<T> task) {
        if (task == null) {
            return null;
        }
        return getThreadExecutor().submit(task);
    }

    /**
     * execute
     * <ul>注：不能保持执行顺序</ul>
     *
     * @param task
     */
    public static void execute(Runnable task) {
        if (task == null) {
            return;
        }
        getThreadExecutor().submit(task);
    }

    /**
     * execute
     * <ul>注：
     * <li>1、不能保持执行顺序</li>
     * <li>2、作用在针对Runnable任务需要有返回结果的调用</li>
     * </ul>
     *
     * @param task
     * @param result
     */
    public static <T> Future<T> execute(Runnable task, T result) {
        if (task == null) {
            return null;
        }
        return getThreadExecutor().submit(task, result);
    }

    /**
     * 批量执行
     * <ul>特殊提醒：
     * <li>1、不能保持执行顺序</li>
     * <li>2、一旦有任务中断InterruptedException，此批次的所有未完成的任务都将取消</li>
     * </ul>
     *
     * @param tasks
     */
    public static <T> List<Future<T>> batchExecute(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        List<Future<T>> futures = null;
        if (tasks == null || tasks.isEmpty()) {
            futures = new ArrayList<Future<T>>(0);
        } else {
            futures = getThreadExecutor().invokeAll(tasks);
        }
        return futures;
    }

    /**
     * 批量执行
     * <ul>特殊提醒：
     * <li>1、不能保持执行顺序</li>
     * <li>2、超过时限后，任何尚未完成的任务都会被取消。(任务要么正常地完成，要么被取消。)</li>
     * <li>3、一旦有任务中断InterruptedException，此批次的所有未完成的任务都会被取消</li>
     * </ul>
     *
     * @param tasks
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    public static <T> List<Future<T>> batchExecute(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        List<Future<T>> futures = null;
        if (tasks == null || tasks.isEmpty()) {
            futures = new ArrayList<Future<T>>(0);
        } else {
            futures = getThreadExecutor().invokeAll(tasks, timeout, unit);
        }
        return futures;
    }


    /**
     * 任务
     */
    public static abstract class Job {
        @Getter
        private Object[] params;

        public Job(Object... params) {
            this.params = params;
        }

        /**
         * 业务处理
         *
         * @param params
         */
        public abstract void process(Object... params);
    }


    /**
     * 执行任务线程
     */
    private static class JobThread implements Runnable {
        private Job job;
        private CountDownLatch latch;

        public JobThread(Job job, CountDownLatch latch) {
            this.job = job;
            this.latch = latch;
        }

        public void run() {
            try {
                this.job.process(job.getParams());
            } finally {
                this.latch.countDown();
            }
        }
    }


    /**
     * <ul>
     * 批量多线程执行多个job，之间中断线程，所有job全部执行完成后才唤醒线程
     * <li>如果不需要保持顺序,采用多线程并行执行的模式，当所有线程都执行完成后才返回<li/>
     * </ul>
     * <pre>调用案例 {@code
     * List<ThreadHelper.Job> jobs = new ArrayList();
     * jobs.add(new ThreadHelper.Job(kafkaRecord) {
     *     @Override
     *     public void process(Object... params) {
     *         //业务处理......
     *     }
     * });
     * ThreadHelper.batchExecute(jobs);
     * }</pre>
     *
     * @param jobs
     */
    public static void batchExecute(List<ThreadHelper.Job> jobs) {
        if (jobs == null) {
            return;
        }
        CountDownLatch latch = new CountDownLatch(jobs.size());
        for (Job job : jobs) {
            ThreadHelper.getThreadExecutor().execute(new JobThread(job, latch));
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
        }
    }
}